package io.netty.bootstrap;

import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

import org.ala.tiktools.rpc.kcp.netty.server.KcpServerChannel;
import org.ala.tiktools.rpc.kcp.netty.server.KcpServerChildChannel;

import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoop;
import io.netty.util.AttributeKey;
import io.netty.util.internal.ObjectUtil;
import io.netty.util.internal.logging.InternalLogger;
import io.netty.util.internal.logging.InternalLoggerFactory;


/**
 * kcp服务端的Bootstrap
 * <p>	抄ServerBootstrap，只不过当有新channel注册进来时，绑的是boos线程
 * <p>	新channel是个虚拟channel，底层没有socket套接字无法收发消息，所以只能由boss线程的udpChannel来读写
 * <p>	每个boss线程对应一个 udp socket，负责数据读写
 * <p>	提供bind(int[] ports)方法，同时绑定多个udp socket 缓解单boss线程压力
 * 
 * @author ala
 * @date 2025-04-14 13:47
 */
public class KcpServerBootstrap extends AbstractBootstrap<KcpServerBootstrap, KcpServerChannel> implements Cloneable {
    private static final InternalLogger logger = InternalLoggerFactory.getInstance(ServerBootstrap.class);

    private final Map<ChannelOption<?>, Object> childOptions = new LinkedHashMap<ChannelOption<?>, Object>();
    private final Map<AttributeKey<?>, Object> childAttrs = new ConcurrentHashMap<AttributeKey<?>, Object>();
    private final KcpServerBootstrapConfig config = new KcpServerBootstrapConfig(this);
    private volatile ChannelHandler childHandler;

    public KcpServerBootstrap() {
        super();
    }
    public KcpServerBootstrap(KcpServerBootstrap bootstrap) {
        super(bootstrap);
        childHandler = bootstrap.childHandler;
        synchronized (bootstrap.childOptions) {
            childOptions.putAll(bootstrap.childOptions);
        }
        childAttrs.putAll(bootstrap.childAttrs);
    }
    
    
    /**
     * 	绑定多端口（同步等待绑定结果）
     * 	<p>	有一个绑定失败则整体失败
     */
	public CompletableFuture<KcpServerChannel[]> binds(int[] inetPorts) {
		CompletableFuture<KcpServerChannel[]> cf = new CompletableFuture<>();
		KcpServerChannel[] channels = new KcpServerChannel[inetPorts.length];
		AtomicInteger succCnt = new AtomicInteger();
		for (int i = 0 ; i < inetPorts.length ; i++) {
			int port = inetPorts[i];
			final int _i = i;
			try {
				ChannelFuture _cf = bind(port);
				_cf.addListener(c -> {
					if (!c.isSuccess()) {
						cf.completeExceptionally(c.cause());
					} else {
						KcpServerChannel channel = (KcpServerChannel) _cf.channel();
						channels[_i] = channel;
						if (succCnt.incrementAndGet() == inetPorts.length) {
							cf.complete(channels);
						}
					}
				});
			} catch (Exception e) {
				cf.completeExceptionally(e);
			}
		}
		return cf;
	}


    /**
     * 初始化KcpServerChannel
     * @param channel   KcpServerChannel
     * @throws Exception
     */
    @Override
    void init(Channel channel) throws Exception {
        KcpServerChannel serverChannel = (KcpServerChannel) channel;

        //  ServerBootstrap源码照抄，不同的是udp不需要建立链接，所以不需要用worker线程轮训新建的channel
        setChannelOptions(serverChannel, options0().entrySet().toArray(newOptionArray(0)), logger);
        setAttributes(serverChannel, attrs0().entrySet().toArray(newAttrArray(0)));

        ChannelPipeline p = serverChannel.pipeline();

        final ChannelHandler currentChildHandler = childHandler;
        final Map.Entry<ChannelOption<?>, Object>[] currentChildOptions = childOptions.entrySet().toArray(newOptionArray(0));
        final Map.Entry<AttributeKey<?>, Object>[] currentChildAttrs = childAttrs.entrySet().toArray(newAttrArray(0));

        p.addLast(new ChannelInitializer<Channel>() {
            @Override
            public void initChannel(final Channel ch) {
                final ChannelPipeline pipeline = ch.pipeline();
                ChannelHandler handler = config.handler();
                if (handler != null) {
                    pipeline.addLast(handler);
                }
                //  udp是无连接的，无需用worker线程去扫socket（也没channel给你扫，KcpClientChannel是个虚拟的channel，底层不对应某个具体的socket套接字）
                ch.eventLoop().execute(new Runnable() {
                    @Override
                    public void run() {
                        //  绑
                        pipeline.addLast(new KcpServerBootstrapAcceptor(ch, currentChildHandler, currentChildOptions, currentChildAttrs));
                    }
                });
            }
        });
    }

    
	/**
     * 追加childHandler
     */
    public KcpServerBootstrap childHandler(ChannelHandler childHandler) {
        this.childHandler = ObjectUtil.checkNotNull(childHandler, "childHandler");
        return this;
    }
    /**
     * 追加childOptions
     */
    final Map<ChannelOption<?>, Object> childOptions() {
        synchronized (childOptions) {
            return copiedMap(childOptions);
        }
    }
    public <T> KcpServerBootstrap childAttr(AttributeKey<T> childKey, T value) {
        ObjectUtil.checkNotNull(childKey, "childKey");
        if (value == null) {
            childAttrs.remove(childKey);
        } else {
            childAttrs.put(childKey, value);
        }
        return this;
    }
    @Override
    public KcpServerBootstrap clone() {
        return new KcpServerBootstrap(this);
    }
    @Override
    public AbstractBootstrapConfig<KcpServerBootstrap, KcpServerChannel> config() {
        return config;
    }


    /**
     *  初始化子channel
     */
    public class KcpServerBootstrapAcceptor extends ChannelInboundHandlerAdapter {
        Channel channel;
        ChannelHandler childHandler;
        Map.Entry<ChannelOption<?>, Object>[] childOptions;
        Map.Entry<AttributeKey<?>, Object>[] childAttrs;

        public KcpServerBootstrapAcceptor(
                Channel ch,
                ChannelHandler currentChildHandler,
                Map.Entry<ChannelOption<?>, Object>[] currentChildOptions,
                Map.Entry<AttributeKey<?>, Object>[] currentChildAttrs) {
            channel = ch;
            childHandler = currentChildHandler;
            childOptions = currentChildOptions;
            childAttrs = currentChildAttrs;
        }

        /**
         *  当有消息过来时
         */
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            //  KcpServerChildChannel是在KcpServerChannel读到数据后自动创建的
            //  自动创建后会手动触发一次fireChannelRead，给的参数就是KcpServerChildChannel
            //  通过这种机制走到这里来的
            Channel parent = ctx.channel();
            KcpServerChildChannel child = (KcpServerChildChannel) msg;

            child.pipeline().addLast(childHandler);
            setChannelOptions(child, childOptions, logger);
            setAttributes(channel, childAttrs);

            //  绑EventLoop
            EventLoop eventLoop = parent.eventLoop();
//            EventLoop eventLoop = workerGroup.next();
            eventLoop.register(child).addListener(cf -> {
                if (!cf.isSuccess()) {
                    forceClose(child, cf.cause());
                } else {
                    //  如果注册成功，执行afterRegister回调
                    child.afterEventLoopRegister();
                }
            });
        }
        public void forceClose(Channel child, Throwable t) {
            child.unsafe().closeForcibly();
            logger.error("accepted channel 注册失败: {}", child, t);
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            super.exceptionCaught(ctx, cause);
        }
    }
}
